package com.qudian.pay.mq.consumer.impl;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import com.qudian.pay.mq.MQException;
import com.qudian.pay.mq.consumer.AbstractMQConsumer;
import com.qudian.pay.mq.enums.MQRuntimeFailStatus;

/**
 * @Description: 多点广播式MQ消费实现
*
 * @date 2016年6月6日 下午5:51:29
 */ 
public abstract class MQBroadCastConsumer extends AbstractMQConsumer {
	
	protected final static Logger logger = LoggerFactory.getLogger(MQBroadCastConsumer.class);
	
	public void init() {
		checkPropertiesAndSetEnv();
		consumer = new DefaultMQPushConsumer();
		consumer.setNamesrvAddr(namesrvAddr);
		consumer.setInstanceName(instanceName);
		// 一次只处理一条消息，控制高可靠
		consumer.setConsumeMessageBatchMaxSize(1);
		consumer.setConsumerGroup(getGroup());
		consumer.setConsumeThreadMin(consumeThreadMin);
		consumer.setConsumeThreadMax(consumeThreadMax);
		consumer.setPullInterval(0);
		consumer.setMessageModel(MessageModel.BROADCASTING);
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
		// 注册接收的Topic,Tag
		try {
			consumer.subscribe(topic, tagExpress);
			consumer.registerMessageListener(this);
			consumer.start();
			logger.info("rockmq consumer start success! " + topic + " | " + tagExpress);
		} catch (MQClientException e) {
			logger.error("RocketMQ consumer subscribe err. nameserver={" + namesrvAddr + "},consumerGroup={" + getGroup() + "}", e);
			throw new MQException(MQRuntimeFailStatus.MQ_CLIENT_FAILURE, e);
		}
	}
	
}
